package com.xbai.spark.core.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * mapPartitionsWithIndex(func) 算子
  * 作用：类似于mapPartitions，但 func 带有一个整数参数表示分片的索引值，因此在类型为 T 的RDD上运行时，func 的函数类型必须是 (Int, Interator[T]) => Iterator[U]；
  * 案例：创建一个RDD，使每个元素跟所在分区形成一个元组组成一个新的RDD
  *
  * @author xbai
  * @Date 2020/12/28
  */
object Spark04_MapPartitionsWithIndex {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("mapPartitionsWithIndex")
    val sc = new SparkContext(conf)

    val listRDD: RDD[Int] = sc.makeRDD(1 to 10)
    val mapPartitionsWithIndexRDD: RDD[(Int, Int)] = listRDD.mapPartitionsWithIndex {
      case (index, data) =>
        data.map((_, index))
    }
    mapPartitionsWithIndexRDD.collect().foreach(println)
  }
}
